package com.yn.xgame;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.yn.xgame.message.C2GPbRedirectMessage;
import com.yn.xgame.pb.ClientGateProto;
import com.yn.xgame.pb.GateClientProto;
import com.yn.xgame.util.Utils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * MQClient
 * Created by wangboo on 2017/7/11.
 */
public class MQClient implements Runnable {

    private Thread consumeThread;

    interface Handler {
        void handleMessage(ConsumerRecord<String, byte[]> rec);
    }

    private static final Logger logger = LoggerFactory.getLogger(MQClient.class);

    private KafkaProducer<String, byte[]> producer;
    private KafkaConsumer<String, byte[]> consumer;
    private Handler handler;

    public MQClient(Properties prop, Handler handler, String group, String... topic) {
        this(prop, handler, group, ImmutableList.copyOf(topic));
    }
    public MQClient(Properties prop, Handler handler, String group, List<String> topic) {
        this.handler = handler;
        consumer = new KafkaConsumer<>(prop, new StringDeserializer(), new ByteArrayDeserializer());
        producer = new KafkaProducer<>(prop, new StringSerializer(), new ByteArraySerializer());
        List<String> subList = new ArrayList<>(topic.size());
        for (String t : topic) {
            if (t == null || t.equals("null")) continue;
            subList.add(t);
            logger.debug("MqClient group: {}, subscribe: {}", group, t);
        }
        consumer.subscribe(subList);
        consumeThread = new Thread(this);
    }

    public void start() {
        consumeThread.start();
    }

    public void shutdown() {
        producer.close();
        consumer.close();
        consumeThread.interrupt();
    }

    /**
     * 将G2C消息通过gate转发给Client
     */
    public void sendClient(String topic, long sid, GateClientProto.G2CMessage pb) {
        if (logger.isDebugEnabled()) {
            logger.debug("sendClient topic: {}, sid: {}, pb: {}", topic, sid, pb);
        }
        RecordHeader header = new RecordHeader("sid", Utils.getLongByte(sid));
        List<Header> headers = Lists.newArrayList(header);
        ProducerRecord<String, byte[]> rec = new ProducerRecord<>(topic, null,
                MqTags.TO_CLIENT, pb.toByteArray(), headers);
        sendImpl(rec);
    }

    /**
     * 将C2GMessage转发给目标逻辑服
     */
    public void sendPbRedirect(String topic, C2GPbRedirectMessage msg) {
        sendPbRedirect(topic, msg.getSessionId(), msg.getFromGateTopic(), msg.getPb());
    }
    public void sendPbRedirect(String toTopic, long sid, String fromTopic, ClientGateProto.C2GMessage pb) {
        RecordHeader sidHeader = new RecordHeader("sid", Utils.getLongByte(sid));
        RecordHeader fromTopicHeader = new RecordHeader("topic", fromTopic.getBytes(Utils.charset));
        List<Header> headers = Lists.newArrayList(sidHeader, fromTopicHeader);
        ProducerRecord<String, byte[]> rec = new ProducerRecord<>(toTopic, null,
                MqTags.C2G_REDIRECT, pb.toByteArray(), headers);
        if (logger.isDebugEnabled()) {
            logger.debug("sendClient to: {}, sid: {}, msg: {}", toTopic, sid, pb);
        }
        sendImpl(rec);
    }

    /**
     * 服务器内，消息转发
     */
    public void sendInter(String fromTopic, String toTopic, String msgId, Map<String, Object> body) {
        RecordHeader msgIdHeader = new RecordHeader("msgId", msgId.getBytes(Utils.charset));
        RecordHeader classHeader = new RecordHeader("class", MqTags.bean_type_map);
        RecordHeader fromHeader = new RecordHeader("from", fromTopic.getBytes(Utils.charset));
        List<Header> headers = ImmutableList.of(msgIdHeader, classHeader, fromHeader);
        byte[] bin = JSON.toJSONBytes(body);
        if (logger.isDebugEnabled()) {
            logger.debug("sendInter to: {}, msgId: {}, obj: {}", toTopic, msgId, new String(bin, Utils.charset));
        }
        sendInter(toTopic, headers, bin);
    }

    public void sendInter(String fromTopic, String toTopic, String msgId, Object obj) {
        RecordHeader msgIdHeader = new RecordHeader("msgId", msgId.getBytes(Utils.charset));
        RecordHeader classHeader = new RecordHeader("class", obj.getClass().getName().getBytes(Utils.charset));
        RecordHeader fromHeader = new RecordHeader("from", fromTopic.getBytes(Utils.charset));
        List<Header> headers = ImmutableList.of(msgIdHeader, classHeader, fromHeader);
        byte[] bin = JSON.toJSONBytes(obj);
        if (logger.isDebugEnabled()) {
            logger.debug("sendInter to: {}, msgId: {}, obj: {}", toTopic, msgId, new String(bin, Utils.charset));
        }
        sendInter(toTopic, headers, bin);
    }

    public void sendInter(String topic, List<Header> headers, byte[] bin) {
        ProducerRecord<String, byte[]> rec = new ProducerRecord<>(topic, null,
                MqTags.BEAN, bin, headers);
        sendImpl(rec);
    }

    private void sendImpl(ProducerRecord<String, byte[]> rec) {
        try {
            producer.send(rec);
        } catch (Exception e) {
            logger.error("sendInter error: ", e);
        }
    }

    @Override
    public void run() {
        while (true) {
            try {
                ConsumerRecords<String, byte[]> all = consumer.poll(1);
                for (ConsumerRecord<String, byte[]> rec : all) {
                    try {
                        handler.handleMessage(rec);
                    } catch (Throwable e) {
                        logger.error("handleMessage error: ", e);
                    }
                }
            } catch (Exception e) {
                logger.error("error: ", e);
            }

        }
    }

}
